#include <sstream>
#include <iostream>
#include "XBaseCommand.h"
#include "XBaseClient.h"
#include "constants.h"
#include "XStringStream.h"
#include "XBaseProtocol.h"
#include "XBaseError.h"
#include "XUtility.h"
#include "XBaseCommandStruct.h"

using namespace std;

XIMPL_LOGGER(XBaseCommandBase, logger);

const uint32_t XBaseCommandBase::_lib_ver = 1;

XBaseCommandBase::XBaseCommandBase(uint16_t cmd_type)
{
    _request_head = new XBaseProtocolHead;
    _response_head = new XBaseProtocolHead;
    
	_request_head->cmd_type() = cmd_type;
	
	_result = XERROR_FAIL;
	_affected_rows = 0;
    mErrno = XBaseClient::PERR_SUCCESS;
}
XBaseCommandBase::~XBaseCommandBase()
{
    delete _request_head;
    delete _response_head;
}
void XBaseCommandBase::set_table_name(const string& table_name)
{
	mTableName = table_name;
}
int XBaseCommandBase::affected_rows(int affected_rows)
{
	int old_affected_rows = _affected_rows;
	_affected_rows += affected_rows;
	return old_affected_rows;
}	
bool XBaseCommandBase::fatal() 
{
	return !(_result == XBaseClient::PERR_SUCCESS 
		|| _result == XBaseClient::PERR_END_OF_SCAN_RECORD
		|| _result == XBaseClient::PERR_WRITE_TABLET_ERROR
		|| _result == XBaseClient::PERR_FAMILY_NOT_EXISTS
		|| _result == XBaseClient::PERR_GROUP_NOT_EXISTS
		|| _result == XBaseClient::PERR_SCANNER_NOT_EXISTS
		|| _result == XBaseClient::PERR_COLUMN_NAME_WITH_REGEXP); 
}
	
bool XBaseCommandBase::fail() 
{
	return (_result != XERROR_SUCC); 
}
std::string XBaseCommandBase::errorstr()
{
	if ( mErrno == XERROR_SERVER_CONNECT
		|| mErrno == XERROR_SERVER_SEND
		|| mErrno == XERROR_SERVER_RECV ) {

		return (xerror( mErrno ) + mErrstr);
	} else if ( mErrno <= XERROR_MAX ) {
		return xerror( mErrno );
	} else {
		return mErrstr;
	}
}

int XBaseCommandBase::encode_request(char* buffer, size_t buffer_size)
{
	int head_size = encode_head(buffer, buffer_size);
	if (head_size < 0) {
		mErrno = XERROR_BUFFER_ENCODE_OVERFLOW;

		return XERROR_FAIL;
	}
	int body_size = encode_body(buffer+head_size, buffer_size-head_size);
    if ( body_size == XERROR_FAIL ) {
        mErrno = XERROR_BUFFER_ENCODE_OVERFLOW;
        return XERROR_FAIL;
    }else if(body_size == XERROR_CANCEL)
        return XERROR_CANCEL;

	XBaseProtocol::set_body_length(buffer, body_size);

	return (head_size + body_size);
}

int XBaseCommandBase::decode_response(char* buffer, size_t buffer_size)
{	
	int body_size = decode_head(buffer, buffer_size);
	if (body_size < 0) {
		mErrno = XERROR_BUFFER_DECODE_UNDERFLOW;

		return XERROR_FAIL;
	}

	int result = decode_body(buffer+XBaseProtocol::get_head_length(), (std::size_t)(uint32_t)body_size);
	if (result == XERROR_FAIL) {
		mErrno = XERROR_BUFFER_DECODE_UNDERFLOW;
		return XERROR_FAIL;
	}
	mErrno = _result;
	return result;
}

int XBaseCommandBase::encode_head(char* buffer, size_t buffer_size)
{
	uint32_t body_size = 0;
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << *_request_head << body_size)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}

	return encoder.length();
}

int XBaseCommandBase::decode_head(char* buffer, size_t buffer_size)
{
	uint32_t body_size = 0;
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(decoder >> *_response_head >> body_size)) {
		XLOG4CPLUS_DEBUG(logger, "buffer underflow");
		return XERROR_FAIL;
	}

	return body_size;
}

int XBaseCommandBase::encode_body(char* buffer, size_t buffer_size)
{
	return XERROR_FAIL;
}

int XBaseCommandBase::decode_body(char * buffer, size_t buffer_size)
{	
	return XERROR_FAIL;
}

XIMPL_LOGGER(XBaseCommandAlter, logger);

XBaseCommandAlter::XBaseCommandAlter() : XBaseCommandBase(0)
{
}
XBaseCommandAlter::~XBaseCommandAlter()
{}
void XBaseCommandAlter::set_group_name(const  std::string & group_name)
{
	_request_head->cmd_type() = DELETE_GROUP_REQ;
	_group_name = group_name;
}
void XBaseCommandAlter::set_family_name(const  std::string & family_name)
{
	_request_head->cmd_type() = DELETE_FAMILY_REQ;
	_family_name = family_name;
}
void XBaseCommandAlter::drop_group(const  std::string & group_name)
{
	_request_head->cmd_type() = DELETE_GROUP_REQ;
	_group_name = group_name;
}
void XBaseCommandAlter::drop_family(const  std::string & family_name)
{
	_request_head->cmd_type() = DELETE_FAMILY_REQ;
	_family_name = family_name;
}
void XBaseCommandAlter::add_group(const std::string& group_name, uint32_t compress_method, uint32_t cache_rate, uint32_t block_size, uint32_t max_num_sstable, uint32_t max_size_sstable, uint32_t max_size_memtable)
{
	_request_head->cmd_type() = ADD_GROUP_REQ;
	_group.name() = group_name;
	_group.compress_method() = compress_method;
	_group.cache_rate() = cache_rate;
	_group.block_size() = block_size;
	_group.max_num_sstable() = max_num_sstable;
	_group.max_size_sstable() = max_size_sstable;
	_group.max_size_memtable() = max_size_memtable;
}
void XBaseCommandAlter::add_family(const std::string& family_name, const std::string& group_name, uint32_t max_num_version, uint32_t max_ttl, uint32_t dead_time)
{
	_request_head->cmd_type() = ADD_FAMILY_REQ;
	_family.name() = family_name;
	_family.group_name() = group_name;
	_family.max_num_version() = max_num_version;
	_family.max_ttl() = max_ttl;
	_family.dead_time() = dead_time;
}
int XBaseCommandAlter::encode_body(char * buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	
	if (_request_head->cmd_type() == ADD_GROUP_REQ) {
		if (!(encoder << _lib_ver << mTableName << _group)) {
			return XERROR_FAIL;
		}
	}
	else if (_request_head->cmd_type() == DELETE_GROUP_REQ) {
		if (!(encoder << _lib_ver << mTableName << _group_name)) {
			return XERROR_FAIL;
		}
	}
	else if (_request_head->cmd_type() == ADD_FAMILY_REQ) {
		if (!(encoder << _lib_ver << mTableName << _family)) {
			return XERROR_FAIL;
		}
	}
	else if (_request_head->cmd_type() == DELETE_FAMILY_REQ) {
		if (!(encoder << _lib_ver << mTableName << _family_name)) {
			return XERROR_FAIL;
		}
	}
	else {
		XLOG4CPLUS_DEBUG(logger, "ALTER_UNKNOWN");

		return XERROR_FAIL;
	}
	
	return encoder.length();
}
int XBaseCommandAlter::decode_body(char* buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);

	if (_request_head->cmd_type() == ADD_GROUP_REQ) {
		if (!(decoder >> _result >> mTableName >> _group_name_response)) {
			XLOG4CPLUS_DEBUG(logger, "buffer overflow");
			return XERROR_FAIL;
		}
	}
	else if (_request_head->cmd_type() == DELETE_GROUP_REQ) {
		if (!(decoder >> _result >> mTableName >> _group_name_response)) {
			XLOG4CPLUS_DEBUG(logger, "buffer overflow");
			return XERROR_FAIL;
		}
	}
	else if (_request_head->cmd_type() == ADD_FAMILY_REQ) {
		if (!(decoder >> _result >> mTableName >> _family_name_response)) {
			XLOG4CPLUS_DEBUG(logger, "buffer overflow");
			return XERROR_FAIL;
		}
	}
	else if (_request_head->cmd_type() == DELETE_FAMILY_REQ) {
		if (!(decoder >> _result >> mTableName >> _family_name_response)) {
			XLOG4CPLUS_DEBUG(logger, "buffer overflow");
			return XERROR_FAIL;
		}
	}
	else {
		XLOG4CPLUS_DEBUG(logger, "ALTERRESP_UNKNOWN");

		return XERROR_FAIL;
	}	

	return _result;
}
std::string XBaseCommandAlter::to_string()
{
	std::string out;

	if (_request_head->cmd_type() == ADD_GROUP_REQ) {
		out += "\n[ALTER_TABLE_ADD_GROUP] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
		out += "\n[ALTER_TABLE_ADD_GROUP] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_ADD_GROUP] group";
		out += _group.to_string();
		
		out += "\n[ALTER_TABLE_ADD_GROUP_RESP] result=";
		out += XUtility::int32_2_string(_result);
		out += "\n[ALTER_TABLE_ADD_GROUP_RESP] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_ADD_GROUP_RESP] group_name=";
		out += _group_name_response;
	}
	else if (_request_head->cmd_type() == DELETE_GROUP_REQ) {
		out += "\n[ALTER_TABLE_DROP_GROUP] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
		out += "\n[ALTER_TABLE_DROP_GROUP] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_DROP_GROUP] group_name=";
		out += _group_name;
		
		out += "\n[ALTER_TABLE_DROP_GROUP_RESP] result=";
		out += XUtility::int32_2_string(_result);
		out += "\n[ALTER_TABLE_DROP_GROUP_RESP] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_DROP_GROUP_RESP] group_name=";
		out += _group_name_response;
	}
	else if (_request_head->cmd_type() == ADD_FAMILY_REQ) {
		out += "\n[ALTER_TABLE_ADD_FAMILY] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
		out += "\n[ALTER_TABLE_ADD_FAMILY] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_ADD_FAMILY] family";
		out += _family.to_string();
		
		out += "\n[ALTER_TABLE_ADD_FAMILY_RESP] result=";
		out += XUtility::int32_2_string(_result);
		out += "\n[ALTER_TABLE_ADD_FAMILY_RESP] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_ADD_FAMILY_RESP] family_name=";
		out += _family_name_response;
	}
	else if (_request_head->cmd_type() == DELETE_FAMILY_REQ) {
		out += "\n[ALTER_TABLE_DROP_FAMILY] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
		out += "\n[ALTER_TABLE_DROP_FAMILY] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_DROP_FAMILY] family_name=";
		out += _family_name;
		
		out += "\n[ALTER_TABLE_DROP_FAMILY_RESP] result=";
		out += XUtility::int32_2_string(_result);
		out += "\n[ALTER_TABLE_DROP_FAMILY_RESP] table_name=";
		out += mTableName;
		out += "\n[ALTER_TABLE_DROP_FAMILY_RESP] family_name=";
		out += _family_name_response;
	}
	else {
		out += "\n[ALTER_TABLE_UNKNOWN] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
	}
	
	return out;
}

XIMPL_LOGGER(XBaseCommandCreate, logger);

XBaseCommandCreate::XBaseCommandCreate() : XBaseCommandBase(CREATE_TABLE_REQ)
{	
}
XBaseCommandCreate::~XBaseCommandCreate()
{
}
void XBaseCommandCreate::add_group(const std::string& group_name, uint32_t compress_method, uint32_t cache_rate, uint32_t block_size, uint32_t max_num_sstable, uint32_t max_size_sstable, uint32_t max_size_memtable)
{
	_group_list.add( group_name, compress_method, cache_rate, block_size, max_num_sstable, max_size_sstable, max_size_memtable);
}

void XBaseCommandCreate::add_family(const std::string& family_name, const std::string& group_name, uint32_t max_num_version, uint32_t max_ttl, uint32_t dead_time)
{
	_family_list.add( family_name, group_name, max_num_version, max_ttl, dead_time);
}

int XBaseCommandCreate::encode_body(char* buffer, size_t buffer_size)
{	
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver << mTableName << _group_list << _family_list)) {
		mErrno = XERROR_BUFFER_ENCODE_OVERFLOW;
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}
	
	return encoder.length();
}

int XBaseCommandCreate::decode_body(char* buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(decoder >> _result >> mTableName)) {
		XLOG4CPLUS_DEBUG(logger, "buffer overflow");
		return XERROR_FAIL;
	}

	return _result;
}

std::string XBaseCommandCreate::to_string()
{
	std::string out;

	out += "\n[CREATE_TABLE] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	out += "\n[CREATE_TABLE] table_name=";
	out += mTableName;
	out += "\n[CREATE_TABLE] ";
	out += _group_list.to_string();
	out += "\n[CREATE_TABLE] ";
	out += _family_list.to_string();
	
	out += "\n[CREATE_TABLE_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[CREATE_TABLE_RESP] table_name=";
	out += mTableName;	

	return out;
}

bool XBaseCommandCreate::is_syntax_valid()
{
	XLOG4CPLUS_GUARD( logger, __FUNCTION__ );
	
	bool res = true;
	
	for ( XBaseFamily::iterator it_f = _family_list.begin(); it_f != _family_list.end(); ++it_f ) {
		XBaseGroup::iterator it_g = _group_list.begin();
		for ( ; it_g != _group_list.end(); ++it_g ) {
			if ( it_g->name() == it_f->group_name() ) {
				break;
			}
		}
		if ( it_g == _group_list.end() ) {
			res = false;
			break;
		}
	}

	return res;
}

XIMPL_LOGGER(XBaseCommandDesc, logger);

XBaseCommandDesc::XBaseCommandDesc() : XBaseCommandBase(TABLE_INFO_REQ)
{	
}
XBaseCommandDesc::~XBaseCommandDesc()
{
}
int XBaseCommandDesc::encode_body(char* buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver << mTableName)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}

	return encoder.length();
}

int XBaseCommandDesc::decode_body(char* buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(decoder >> _result >> mTableName >> _group >> _family)) {
		XLOG4CPLUS_DEBUG(logger, "buffer overflow");
		return XERROR_FAIL;
	}

	return _result;
}

std::string XBaseCommandDesc::to_string()
{
	std::string out;

	out += "\n[DESC_TABLE] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	out += "\n[DESC_TABLE] table_name=";
	out += mTableName;

	out += "\n[DESC_TABLE_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[DESC_TABLE_RESP] table_name=";
	out += mTableName;
	out += "\n[DESC_TABLE_RESP] ";
	out += _group.to_string();
	out += "\n[DESC_TABLE_RESP] ";
	out += _family.to_string();

	return out;
}

XIMPL_LOGGER(XBaseCommandDrop, logger);

XBaseCommandDrop::XBaseCommandDrop() : XBaseCommandBase(DELETE_TABLE_REQ)
{	
}
XBaseCommandDrop::~XBaseCommandDrop()
{
}
int XBaseCommandDrop::encode_body(char* buffer, size_t buffer_size)
{	
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver << mTableName)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}
	
	return encoder.length();
}

int XBaseCommandDrop::decode_body(char* buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(decoder >> _result >> mTableName)) {
		XLOG4CPLUS_DEBUG(logger, "buffer overflow");
		return XERROR_FAIL;
	}

	return _result;
}

std::string XBaseCommandDrop::to_string()
{
	std::string out;

	out += "\n[DROP_TABLE] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	out += "\n[DROP_TABLE] table_name=";	
	out += mTableName;	
	
	out += "\n[DROP_TABLE_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[DROP_TABLE_RESP] table_name=";
	out += mTableName;

	return out;
}

XIMPL_LOGGER(XBaseCommandInsert, logger);

XBaseCommandInsert::XBaseCommandInsert() : XBaseCommandBase(WRITE_REQ)
{	
	_affected_rows = 0;
}
XBaseCommandInsert::~XBaseCommandInsert()
{
}
void XBaseCommandInsert::add(const std::string& value, const std::string& row, const std::string& column, uint64_t timestamp)
{
	mRow = row;
	
	Record record;
	record.key().row() = row;
	record.key().column() = column;
	record.key().timestamp() = timestamp;
	record.value() = value;

	if ( !record.key().column().empty() && record.key().column() != ANY_COLUMN && record.key().column().find(':') == std::string::npos) {
		record.key().column() += ":";
	}

	_record._list.push_back(record);
}

int XBaseCommandInsert::encode_body(char * buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver << mTabletName << _record)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}
	
	return encoder.length();
}

int XBaseCommandInsert::decode_body(char * buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	
	if (!(decoder >> _result >> mTabletName >> _affected_rows)) {
		XLOG4CPLUS_DEBUG(logger, "buffer underflow");
		return XERROR_FAIL;
	}

	return _result;
}

std::string XBaseCommandInsert::to_string()
{
	std::string out;

	out += "\n[INSERT_RECORD] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	out += "\n[INSERT_RECORD] table_name=";	
	out += mTableName;
	out += "\n[INSERT_RECORD] tablet_name=";	
	out += mTabletName;
	out += "\n[INSERT_RECORD] ";
	out += _record.to_string();
	
	out += "\n[INSERT_RECORD_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[INSERT_RECORD_RESP] tablet_name=";
	out += mTabletName;	
	out += "\n[INSERT_RECORD_RESP] affected_rows=";
	out += XUtility::uint32_2_string(_affected_rows);

	return out;
}

//DoZerg
XIMPL_LOGGER(XBaseCommandBatchInsert, logger);

XBaseCommandBatchInsert::XBaseCommandBatchInsert()
    : XBaseCommandBase(BATCH_WRITE_REQ)
    , record_count_(0)
{}

bool XBaseCommandBatchInsert::remove_finished()
{
    __Row::iterator it = _row.begin();
    for(size_t i = 0;i < _status_by_row.size() && it != _row.end();++i){
        if(_status_by_row[i] == XBaseClient::PERR_SUCCESS){
            record_count_ -= it->second.size();
            _row.erase(it++);
        }else
            ++it;
    }
    _status_by_row.clear();
    _result = 0;
    mErrno = 0;
    mErrstr.clear();
    return !_row.empty();
}

int XBaseCommandBatchInsert::encode_body(char * buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if(!(encoder<<_lib_ver<<mTabletName)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}
    size_t row_pos = encoder.length();
    if(!(encoder<<U32(_row.size()))){
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
    }
    __Row::iterator i,end;
    i = (start_row_.empty() ? _row.begin() : _row.lower_bound(start_row_));
    end = (end_row_.empty() ? _row.end() : _row.lower_bound(end_row_));
    U32 row_count = 0;
    for(;i != _row.end() && i != end;){
        if(i->second.empty()){
            _row.erase(i++);
            continue;
        }
        uint32_t cur = encoder.length();
        if(!(encoder<<cur<<i->second)){
		    XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		    return XERROR_FAIL;
        }
        uint32_t now = encoder.length();
        encoder.lseek(cur);
        encoder<<(now - cur - 4);
        encoder.lseek(now);
        ++row_count;
        ++i;
    }
    if(!row_count)
        return XERROR_CANCEL;
    size_t now = encoder.length();
    encoder.lseek(row_pos);
    encoder<<row_count;
    encoder.lseek(now);
#if 0
    cout<<"----------------------\n"
        <<XUtility::str_2_hex(buffer,encoder.length())
        <<"\n----------------------";
#endif
	return encoder.length();
}

int XBaseCommandBatchInsert::decode_body(char * buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	uint32_t i = 0;
	if (!(decoder>>_result>>mTabletName>>i)) {
		XLOG4CPLUS_DEBUG(logger, "buffer underflow");
		return XERROR_FAIL;
	}
    _status_by_row.resize(i + _status_by_row.size());
    for(i = _status_by_row.size() - i;i < _status_by_row.size();++i){
        if(!(decoder>>_status_by_row[i])){
            XLOG4CPLUS_DEBUG(logger, "buffer underflow");
            return XERROR_FAIL;
        }
    }
	return _result;
}

std::string XBaseCommandBatchInsert::to_string()
{
	std::ostringstream oss;
	oss<<"\n[BAT_INSERT_RECORD] libv="<<_lib_ver
	    <<"\n[BAT_INSERT_RECORD] tablet_name="<<mTabletName
	    <<"\n[BAT_INSERT_RECORD] row_count="<<_row.size();
    size_t index = 0;

	oss<<"\n[BAT_INSERT_RECORD_RESP] result="<<_result
	    <<"\n[BAT_INSERT_RECORD_RESP] tablet_name="<<mTabletName
	    <<"\n[BAT_INSERT_RECORD_RESP] affected_rows="<<_affected_rows
	    <<"\n[BAT_INSERT_RECORD_RESP] status_number="<<_status_by_row.size();
    for(index = 0;index < _status_by_row.size() && index < 10;++index)
        oss<<"\n[BAT_INSERT_RECORD_RESP] status["<<index<<"]="<<_status_by_row[index];
    if(index < _status_by_row.size())
        oss<<"\n...";
	return oss.str();
}

XIMPL_LOGGER(XBaseCommandList, logger);

XBaseCommandList::XBaseCommandList() : XBaseCommandBase(LIST_TABLET_REQ)
{	
}
XBaseCommandList::~XBaseCommandList()
{}

int XBaseCommandList::encode_body(char* buffer, size_t buffer_size)
{	
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}
	
	return encoder.length();
}

int XBaseCommandList::decode_body(char* buffer, size_t buffer_size)
{
	uint32_t tabletNum;
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(decoder >> _result >> tabletNum)) {
		XLOG4CPLUS_DEBUG(logger, "buffer overflow");
		return XERROR_FAIL;
	}
	for (uint32_t i=0; i<tabletNum; i++) {
		TabletInfo tabletInfo;
		if ( !(decoder >> tabletInfo) ) {
			XLOG4CPLUS_DEBUG(logger, "buffer overflow");
			return XERROR_FAIL;
		}
		mTabletInfoList.push_back( tabletInfo );
	}

	return _result;
}

std::string XBaseCommandList::to_string()
{
	std::string out;

	out += "\n[LIST_TABLET] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	
	out += "\n[LIST_TABLET_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[LIST_TABLET_RESP] tablet=";
	for ( std::vector<TabletInfo>::iterator it = mTabletInfoList.begin(); it != mTabletInfoList.end(); ++it ) {
		out += it->to_string();
		out += ",";
	}

	return out;
}

XIMPL_LOGGER(XBaseCommandScan, logger);

XBaseCommandScan::XBaseCommandScan() : XBaseCommandBase(0)
{	
	_state = SCAN_UNKNOWN;
	_scaner_id_request = 0;
	_scaner_id_response = 0;
	mLatestTimestamp = 0;
	mOldestTimestamp = 0;
	mTtl = 0;
	
	_timestamp = 0;
	_version = 1;
	_option = 0;
	mNumRow = 3;
	mIsAlive = false;
}
XBaseCommandScan::XBaseCommandScan(const std::string& tablet_name, const std::string& row_start, const std::string& row_end, const std::string& column, uint64_t timestamp, uint32_t version) : XBaseCommandBase(0)
{	
	_state = SCAN_UNKNOWN;
	_scaner_id_request = 0;
	_scaner_id_response = 0;
	mLatestTimestamp = 0;
	mOldestTimestamp = 0;
	mTtl = 0;
	_option = 0;
	mNumRow = 3;
	mIsAlive = false;

	mTabletName = tablet_name;
	_start_row = row_start;
	_end_row = row_end;
	_column = column;
	mColumnList.push_back( column );
	_timestamp = timestamp;
	_version = version;
}
XBaseCommandScan::~XBaseCommandScan()
{
}
void XBaseCommandScan::set_tablet_name(const std::string& tablet_name)
{
	mTabletName = tablet_name;
}
void XBaseCommandScan::set_row(const std::string& start_row, const std::string& end_row)
{
	mRow = start_row;
	_start_row = start_row;
	_end_row = end_row;
}

void XBaseCommandScan::set_column(const std::string&  column)
{
	_column = column;

	if (  !_column.empty() &&  _column != ANY_COLUMN && _column.find(':') == std::string::npos) {
		_column += ":";
	}
	mColumnList.push_back( _column );
}

void XBaseCommandScan::add_column(const std::string&  column)
{
	string column_tmp = column;

	if (  !column_tmp.empty() &&  column_tmp != ANY_COLUMN && column_tmp.find(':') == std::string::npos) {
		column_tmp += ":";
	}

	mColumnList.push_back( column_tmp );
}

void XBaseCommandScan::set_timestamp(uint64_t timestamp)
{
	_timestamp = timestamp;
}

void XBaseCommandScan::set_latesttimestamp(uint64_t timestamp)
{
	mLatestTimestamp = timestamp;
}

void XBaseCommandScan::set_oldesttimestamp(uint64_t timestamp)
{
	mOldestTimestamp = timestamp;
}

void XBaseCommandScan::set_ttl(uint32_t ttl)
{
	mTtl = ttl;
}

void XBaseCommandScan::set_version(uint32_t version)
{
	_version = version;
}
void XBaseCommandScan::set_option(uint32_t option)
{
	_option = option;
}
void XBaseCommandScan::set_num_row(uint32_t num_row)
{
	mNumRow = num_row;
}
int XBaseCommandScan::encode_body(char * buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
		
	if (_state == SCAN_NEXT) {
		XLOG4CPLUS_DEBUG(logger, "SCAN_NEXT");
		
		if (!(encoder << _lib_ver << mTabletName << _scaner_id_request)) {
			XLOG4CPLUS_DEBUG(logger, "not enough buffer");
			return XERROR_FAIL;
		}
	}
	else if (_state == SCAN_OPEN) {
		XLOG4CPLUS_DEBUG(logger, "SCAN_OPEN");

		encoder << _lib_ver << mTabletName << _start_row << _end_row << U32(mColumnList.size());
		for ( vector<string>::iterator it = mColumnList.begin(); it != mColumnList.end(); ++it ) {
			encoder << *it;
		}
		//if (!(encoder << _lib_ver << mTabletName << _start_row << _end_row << _column << _timestamp << _version << _option << mNumRow)) {
		if (!(encoder << mLatestTimestamp << mOldestTimestamp << mTtl << _version << _option << mNumRow)) {
			XLOG4CPLUS_DEBUG(logger, "not enough buffer");
			return XERROR_FAIL;
		}
	}
	else if (_state == SCAN_CLOSE) {
		XLOG4CPLUS_DEBUG(logger, "SCAN_CLOSE");

		if (!(encoder << _lib_ver << mTabletName << _scaner_id_request)) {
			XLOG4CPLUS_DEBUG(logger, "not enough buffer");
			return XERROR_FAIL;
		}
	}
	else {
		XLOG4CPLUS_DEBUG(logger, "SCAN_UNKNOWN");

		return XERROR_FAIL;
	}

	return encoder.length();
}

int XBaseCommandScan::decode_body(char * buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);

	if (_state == SCAN_NEXT) {
		XLOG4CPLUS_DEBUG(logger, "SCAN_NEXT_RESP");
		
		if (!(decoder >> _result >> _scaner_id_request >> mTabletName >> _record)) {
			XLOG4CPLUS_DEBUG(logger, "buffer underflow");
			return XERROR_FAIL;
		}
	}
	else if (_state == SCAN_OPEN) {
		XLOG4CPLUS_DEBUG(logger, "SCAN_OPEN_RESP");

		if (!(decoder >> _result >> mTabletName >> _scaner_id_response)) {
			XLOG4CPLUS_DEBUG(logger, "buffer underflow");
			return XERROR_FAIL;
		}
		_scaner_id_request = _scaner_id_response;
	}
	else if (_state == SCAN_CLOSE) {
		XLOG4CPLUS_DEBUG(logger, "SCAN_CLOSE_RESP");

		if (!(decoder >> _result >> mTabletName >> _scaner_id_response)) {
			XLOG4CPLUS_DEBUG(logger, "buffer underflow");
			return XERROR_FAIL;
		}
	}
	else {
		XLOG4CPLUS_DEBUG(logger, "SCAN_UNKNOWN");

		return XERROR_FAIL;
	}

	return _result;
}

void XBaseCommandScan::open()
{
	_request_head->cmd_type() = SCAN_OPEN_REQ;
	_state = SCAN_OPEN;
}

void XBaseCommandScan::close()
{
	_result = XERROR_FAIL;
	_request_head->cmd_type() = SCAN_CLOSE_REQ;
	_state = SCAN_CLOSE;
}

void XBaseCommandScan::next()
{
	_result = XERROR_FAIL;
	_record._list.clear();
	_request_head->cmd_type() = SCAN_NEXT_REQ;
	_state = SCAN_NEXT;
}

std::string XBaseCommandScan::to_string()
{
	std::string out;

	if (_state == SCAN_NEXT) {
		out += "\n[SCAN_NEXT] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
		out += "\n[SCAN_NEXT] tablet_name=";
		out += mTabletName;
		out += "\n[SCAN_NEXT] scan_id=";
		out += XUtility::uint64_2_string(_scaner_id_request);

		out += "\n[SCAN_NEXT_RESP] result=";
		out += XUtility::int32_2_string(_result);
		out += "\n[SCAN_NEXT_RESP] table_name=";
		out += mTabletName;
		out += "\n[SCAN_NEXT_RESP] scan_id=";
		out += XUtility::uint64_2_string(_scaner_id_response);
		out += "\n[SCAN_NEXT_RESP] ";
		out += _record.to_string();
	}
	else if (_state == SCAN_OPEN) {
		out += "\n[SCAN_OPEN] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
		out += "\n[SCAN_OPEN] tablet_name=";
		out += mTabletName;
		out += "\n[SCAN_OPEN] start_row=";
		out += _start_row;
		out += "\n[SCAN_OPEN] end_row=";
		out += _end_row;
		out += "\n[SCAN_OPEN] column[]=";
		for (vector<string>::iterator it = mColumnList.begin(); it != mColumnList.end(); ++it) {
			out += "\"";
			out += *it;
			out += "\" ";
		}
		out += "\n[SCAN_OPEN] latesttimestamp=";
		out += XUtility::uint64_2_string(mLatestTimestamp);
		out += "\n[SCAN_OPEN] oldesttimestamp=";
		out += XUtility::uint64_2_string(mOldestTimestamp);
		out += "\n[SCAN_OPEN] ttl=";
		out += XUtility::uint32_2_string(mTtl);
		out += "\n[SCAN_OPEN] version=";
		out += XUtility::uint32_2_string(_version);
		out += "\n[SCAN_OPEN] option=";
		out += XUtility::uint32_2_string(_option);
		out += "\n[SCAN_OPEN] num_row=";
		out += XUtility::uint32_2_string(mNumRow);
		
		out += "\n[SCAN_OPEN_RESP] result=";
		out += XUtility::int32_2_string(_result);
		out += "\n[SCAN_OPEN_RESP] table_name=";
		out += mTabletName;
		out += "\n[SCAN_OPEN_RESP] scan_id=";
		out += XUtility::uint64_2_string(_scaner_id_response);
	}
	else if (_state == SCAN_CLOSE) {
		out += "\n[SCAN_CLOSE] libv=";
		out += XUtility::uint32_2_string(_lib_ver);
		out += "\n[SCAN_CLOSE] tablet_name=";
		out += mTabletName;
		out += "\n[SCAN_CLOSE] scan_id=";
		out += XUtility::uint64_2_string(_scaner_id_request);

		out += "\n[SCAN_CLOE_RESP] result=";
		out += XUtility::int32_2_string(_result);
		out += "\n[SCAN_CLOE_RESP] table_name=";
		out += mTabletName;
		out += "\n[SCAN_CLOE_RESP] scan_id=";
		out += XUtility::uint64_2_string(_scaner_id_response);
	}
	else {
		XLOG4CPLUS_DEBUG(logger, "SCAN_UNKNOWN");
	}	

	return out;
}
bool XBaseCommandScan::fail()
{
	return (_result != XERROR_SUCC && _result != XBaseClient::PERR_END_OF_SCAN_RECORD);
}
bool XBaseCommandScan::eof()
{
	return (_result == XBaseClient::PERR_END_OF_SCAN_RECORD);
}
XIMPL_LOGGER(XBaseCommandSelect, logger);

XBaseCommandSelect::XBaseCommandSelect() : XBaseCommandBase(READ_REQ)
{	
	_key.timestamp() = 0;
}
XBaseCommandSelect::~XBaseCommandSelect()
{
}
void XBaseCommandSelect::set_row(const std::string& row)
{
	mRow = row;
	_key.row() = mRow;
}
void XBaseCommandSelect::set_column(const std::string& column)
{
	_key.column() = column;

	if ( !_key.column().empty() && _key.column() != ANY_COLUMN && _key.column().find(':') == std::string::npos) {
		_key.column() += ":";
	}
}


int XBaseCommandSelect::encode_body(char* buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver << mTabletName << _key)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}
	
	return encoder.length();
}

int XBaseCommandSelect::decode_body(char* buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);

	if (!(decoder >> _result >> mTableName >> _record_list)) {
		XLOG4CPLUS_DEBUG(logger, "buffer underflow");
		return XERROR_FAIL;
	}	
	
	return _result;
}

std::string XBaseCommandSelect::to_string()
{
	std::string out;

	out += "\n[SELECT_RECORD] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	out += "\n[SELECT_RECORD] tablet_name=";
	out += mTabletName;
	out += "\n[SELECT_RECORD] key=";
	out += _key.to_string();

	out += "\n[SELECT_RECORD_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[SELECT_RECORD_RESP] tablet_name=";
	out += mTableName;	
	out += "\n[SELECT_RECORD_RESP] ";
	out += _record_list.to_string();

	return out;
}

XIMPL_LOGGER(XBaseCommandShow, logger);


XBaseCommandShow::XBaseCommandShow() : XBaseCommandBase(LIST_TABLE_REQ)
{	
	mRequestNumber = 0;
}
XBaseCommandShow::~XBaseCommandShow()
{
}
void XBaseCommandShow::set_table_num(int table_num)
{
	mRequestNumber = table_num;
}
int XBaseCommandShow::encode_body(char* buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver << mTableName << mRequestNumber)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}

	return encoder.length();
}

int XBaseCommandShow::decode_body(char* buffer, size_t buffer_size)
{
	uint32_t table_num;
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(decoder >> _result >> table_num)) {
		XLOG4CPLUS_DEBUG(logger, "buffer overflow");
		return XERROR_FAIL;
	}
	for (uint32_t i=0; i<table_num; i++) {
		std::string table_name;
		if ( !(decoder >> table_name) ) {
			XLOG4CPLUS_DEBUG(logger, "buffer overflow");
			return XERROR_FAIL;
		}
		mTableNameList.push_back( table_name );
	}

	return _result;
}

std::string XBaseCommandShow::to_string()
{
	std::string out;

	out += "\n[SHOW_TABLE] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	out += "\n[SHOW_TABLE] table_name=";
	out += mTableName;
	out += "\n[SHOW_TABLE] table_num=";
	out += XUtility::int32_2_string( mRequestNumber );

	out += "\n[SHOW_TABLE_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[SHOW_TABLE_RESP] table_name=";
	for ( std::vector<std::string>::iterator it = mTableNameList.begin(); it != mTableNameList.end(); ++it ) {
		out += *it;
		out += ",";
	}

	return out;
}

XIMPL_LOGGER(XBaseCommandRootTablet, logger);

XBaseCommandRootTablet::XBaseCommandRootTablet() : XBaseCommandBase(ROOT_META_REQ)
{	
}
XBaseCommandRootTablet::~XBaseCommandRootTablet()
{}

 int XBaseCommandRootTablet::encode_body(char* buffer, size_t buffer_size)
{
	XStringStream encoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(encoder << _lib_ver)) {
		XLOG4CPLUS_DEBUG(logger, "not enough buffer");
		return XERROR_FAIL;
	}
	
	return encoder.length();
}

int XBaseCommandRootTablet::decode_body(char* buffer, size_t buffer_size)
{
	XStringStream decoder(buffer, buffer_size, XStringStream::NODELETE, XStringStream::BIGENDIAN);
	if (!(decoder >> _result >> _tablet_info >> _tablet_address)) {
		XLOG4CPLUS_DEBUG(logger, "buffer underflow");
		return XERROR_FAIL;
	}

	return _result;
}

std::string XBaseCommandRootTablet::to_string()
{
	std::string out;
	out += "\n[ROOT_TABLET] libv=";
	out += XUtility::uint32_2_string(_lib_ver);
	
	out += "\n[ROOT_TABLET_RESP] result=";
	out += XUtility::int32_2_string(_result);
	out += "\n[ROOT_TABLET_RESP] tablet_info=";
	out += _tablet_info.to_string();
	out += "\n[ROOT_TABLET_RESP] tablet_address=";
	out += _tablet_address.to_string();

	return out;
}

